{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# dask 大数据处理"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### dask客户端提供一个可视化面板以监控计算过程（新建一个浏览器窗口访问下面的链接）"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table style=\"border: 2px solid white;\">\n",
       "<tr>\n",
       "<td style=\"vertical-align: top; border: 0px solid white\">\n",
       "<h3>Client</h3>\n",
       "<ul>\n",
       "  <li><b>Scheduler: </b>tcp://127.0.0.1:63486\n",
       "  <li><b>Dashboard: </b><a href='http://127.0.0.1:8787/status' target='_blank'>http://127.0.0.1:8787/status</a>\n",
       "</ul>\n",
       "</td>\n",
       "<td style=\"vertical-align: top; border: 0px solid white\">\n",
       "<h3>Cluster</h3>\n",
       "<ul>\n",
       "  <li><b>Workers: </b>1</li>\n",
       "  <li><b>Cores: </b>4</li>\n",
       "  <li><b>Memory: </b>2.00 GB</li>\n",
       "</ul>\n",
       "</td>\n",
       "</tr>\n",
       "</table>"
      ],
      "text/plain": [
       "<Client: scheduler='tcp://127.0.0.1:63486' processes=1 cores=4>"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from dask.distributed import Client\n",
    "client = Client(n_workers=1, threads_per_worker=4, processes=True, memory_limit='2GB')\n",
    "client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import pandas as pd\n",
    "\n",
    "import dask.dataframe as dd\n",
    "import dask.array as da\n",
    "import dask.bag as db"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### dask中的dataframe，2400x2，由10个分区组成，每个分区240行。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div><strong>Dask DataFrame Structure:</strong></div>\n",
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>a</th>\n",
       "      <th>b</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>npartitions=10</th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>2021-09-01 00:00:00</th>\n",
       "      <td>int32</td>\n",
       "      <td>object</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2021-09-11 00:00:00</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>...</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2021-11-30 00:00:00</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2021-12-09 23:00:00</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>\n",
       "<div>Dask Name: from_pandas, 10 tasks</div>"
      ],
      "text/plain": [
       "Dask DataFrame Structure:\n",
       "                         a       b\n",
       "npartitions=10                    \n",
       "2021-09-01 00:00:00  int32  object\n",
       "2021-09-11 00:00:00    ...     ...\n",
       "...                    ...     ...\n",
       "2021-11-30 00:00:00    ...     ...\n",
       "2021-12-09 23:00:00    ...     ...\n",
       "Dask Name: from_pandas, 10 tasks"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "index = pd.date_range(\"2021-09-01\", periods=2400, freq=\"1H\")\n",
    "df = pd.DataFrame({\"a\": np.arange(2400), \"b\": list(\"abcaddbe\" * 300)}, index=index)\n",
    "ddf = dd.from_pandas(df, npartitions=10)\n",
    "ddf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(Timestamp('2021-09-01 00:00:00', freq='H'),\n",
       " Timestamp('2021-09-11 00:00:00', freq='H'),\n",
       " Timestamp('2021-09-21 00:00:00', freq='H'),\n",
       " Timestamp('2021-10-01 00:00:00', freq='H'),\n",
       " Timestamp('2021-10-11 00:00:00', freq='H'),\n",
       " Timestamp('2021-10-21 00:00:00', freq='H'),\n",
       " Timestamp('2021-10-31 00:00:00', freq='H'),\n",
       " Timestamp('2021-11-10 00:00:00', freq='H'),\n",
       " Timestamp('2021-11-20 00:00:00', freq='H'),\n",
       " Timestamp('2021-11-30 00:00:00', freq='H'),\n",
       " Timestamp('2021-12-09 23:00:00', freq='H'))"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ddf.divisions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 索引"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Dask Series Structure:\n",
       "npartitions=10\n",
       "2021-09-01 00:00:00    object\n",
       "2021-09-11 00:00:00       ...\n",
       "                        ...  \n",
       "2021-11-30 00:00:00       ...\n",
       "2021-12-09 23:00:00       ...\n",
       "Name: b, dtype: object\n",
       "Dask Name: getitem, 20 tasks"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ddf.b"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div><strong>Dask DataFrame Structure:</strong></div>\n",
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>a</th>\n",
       "      <th>b</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>npartitions=1</th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>2021-10-01 00:00:00.000000000</th>\n",
       "      <td>int32</td>\n",
       "      <td>object</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2021-10-09 05:00:59.999999999</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>\n",
       "<div>Dask Name: loc, 11 tasks</div>"
      ],
      "text/plain": [
       "Dask DataFrame Structure:\n",
       "                                   a       b\n",
       "npartitions=1                               \n",
       "2021-10-01 00:00:00.000000000  int32  object\n",
       "2021-10-09 05:00:59.999999999    ...     ...\n",
       "Dask Name: loc, 11 tasks"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ddf[\"2021-10-01\": \"2021-10-09 5:00\"]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### dask dataframe 是延迟执行"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>a</th>\n",
       "      <th>b</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>2021-10-01 00:00:00</th>\n",
       "      <td>720</td>\n",
       "      <td>a</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2021-10-01 01:00:00</th>\n",
       "      <td>721</td>\n",
       "      <td>b</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2021-10-01 02:00:00</th>\n",
       "      <td>722</td>\n",
       "      <td>c</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2021-10-01 03:00:00</th>\n",
       "      <td>723</td>\n",
       "      <td>a</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2021-10-01 04:00:00</th>\n",
       "      <td>724</td>\n",
       "      <td>d</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                       a  b\n",
       "2021-10-01 00:00:00  720  a\n",
       "2021-10-01 01:00:00  721  b\n",
       "2021-10-01 02:00:00  722  c\n",
       "2021-10-01 03:00:00  723  a\n",
       "2021-10-01 04:00:00  724  d"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ddf[\"2021-10-01\": \"2021-10-09 5:00\"].compute().head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0    a\n",
       "1    b\n",
       "2    c\n",
       "3    d\n",
       "4    e\n",
       "Name: b, dtype: object"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ddf.a.mean()\n",
    "ddf.a.mean().compute()\n",
    "ddf.b.unique()\n",
    "ddf.b.unique().compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2021-10-01 00:00:00       620\n",
       "2021-10-01 01:00:00      1341\n",
       "2021-10-01 02:00:00      2063\n",
       "2021-10-01 03:00:00      2786\n",
       "2021-10-01 04:00:00      3510\n",
       "2021-10-01 05:00:00      4235\n",
       "2021-10-01 06:00:00      4961\n",
       "2021-10-01 07:00:00      5688\n",
       "2021-10-01 08:00:00      6416\n",
       "2021-10-01 09:00:00      7145\n",
       "2021-10-01 10:00:00      7875\n",
       "2021-10-01 11:00:00      8606\n",
       "2021-10-01 12:00:00      9338\n",
       "2021-10-01 13:00:00     10071\n",
       "2021-10-01 14:00:00     10805\n",
       "2021-10-01 15:00:00     11540\n",
       "2021-10-01 16:00:00     12276\n",
       "2021-10-01 17:00:00     13013\n",
       "2021-10-01 18:00:00     13751\n",
       "2021-10-01 19:00:00     14490\n",
       "2021-10-01 20:00:00     15230\n",
       "2021-10-01 21:00:00     15971\n",
       "2021-10-01 22:00:00     16713\n",
       "2021-10-01 23:00:00     17456\n",
       "2021-10-02 00:00:00     18200\n",
       "2021-10-02 01:00:00     18945\n",
       "2021-10-02 02:00:00     19691\n",
       "2021-10-02 03:00:00     20438\n",
       "2021-10-02 04:00:00     21186\n",
       "2021-10-02 05:00:00     21935\n",
       "                        ...  \n",
       "2021-10-08 00:00:00    135776\n",
       "2021-10-08 01:00:00    136665\n",
       "2021-10-08 02:00:00    137555\n",
       "2021-10-08 03:00:00    138446\n",
       "2021-10-08 04:00:00    139338\n",
       "2021-10-08 05:00:00    140231\n",
       "2021-10-08 06:00:00    141125\n",
       "2021-10-08 07:00:00    142020\n",
       "2021-10-08 08:00:00    142916\n",
       "2021-10-08 09:00:00    143813\n",
       "2021-10-08 10:00:00    144711\n",
       "2021-10-08 11:00:00    145610\n",
       "2021-10-08 12:00:00    146510\n",
       "2021-10-08 13:00:00    147411\n",
       "2021-10-08 14:00:00    148313\n",
       "2021-10-08 15:00:00    149216\n",
       "2021-10-08 16:00:00    150120\n",
       "2021-10-08 17:00:00    151025\n",
       "2021-10-08 18:00:00    151931\n",
       "2021-10-08 19:00:00    152838\n",
       "2021-10-08 20:00:00    153746\n",
       "2021-10-08 21:00:00    154655\n",
       "2021-10-08 22:00:00    155565\n",
       "2021-10-08 23:00:00    156476\n",
       "2021-10-09 00:00:00    157388\n",
       "2021-10-09 01:00:00    158301\n",
       "2021-10-09 02:00:00    159215\n",
       "2021-10-09 03:00:00    160130\n",
       "2021-10-09 04:00:00    161046\n",
       "2021-10-09 05:00:00    161963\n",
       "Freq: H, Name: a, Length: 198, dtype: int32"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "result = ddf[\"2021-10-01\": \"2021-10-09 5:00\"].a.cumsum() - 100\n",
    "result.compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "ename": "ModuleNotFoundError",
     "evalue": "No module named 'graphviz'",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mModuleNotFoundError\u001b[0m                       Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-22-dfa7c3936fe3>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[1;32m----> 1\u001b[1;33m \u001b[1;32mimport\u001b[0m \u001b[0mgraphviz\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m      2\u001b[0m \u001b[0mresult\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mdask\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      3\u001b[0m \u001b[0mresult\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mvisualize\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mModuleNotFoundError\u001b[0m: No module named 'graphviz'"
     ]
    }
   ],
   "source": [
    "import graphviz\n",
    "result.dask\n",
    "result.visualize()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
